package com.baishancloud.log.quality.sink

import cn.hutool.http.HttpUtil
import cn.hutool.json.JSONUtil
import com.baishancloud.log.quality.pojo.RateResult
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext}
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.JavaConverters.mutableSeqAsJavaListConverter
import scala.collection.mutable
import scala.util.control.Breaks

/**
 * @param threshold 发送请求阈值
 * @author ziqiang.wang 
 * @date 2021/11/16 09:53
 */
class HttpPostSink(threshold: Int = Int.MaxValue) extends SinkFunction[RateResult] with CheckpointedFunction {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[HttpPostSink])

  //上批数据时间
  var timestamp: Long = Long.MaxValue
  //算子内数据结构，存放累计的结果数据
  val results: mutable.ListBuffer[RateResult] = mutable.ListBuffer[RateResult]()
  //将未被发送的数据放到该state，以便checkpoint时进行持久化
  var resultsState: ListState[RateResult] = _
  val url: String = "http://127.0.0.1:10699/v2/push"


  override def invoke(value: RateResult, context: SinkFunction.Context): Unit = {
    if (timestamp == Long.MaxValue) {
      timestamp = value.time
    } else if ((value.time > timestamp || results.size >= threshold) && results.nonEmpty) {
      //新一批数据到达，或者是累计数据数量到达设定阈值，则进行post请求发送数据
      Breaks.breakable {
        for (i <- 1 to 3) {
          val response: String = HttpUtil.post(url, JSONUtil.toJsonStr(results.asJava))
          if (response.contains("OK")) {
            results.clear()
            Breaks.break()
          } else {
            LOGGER.error(s"post请求错误，进行第${i}次重试，错误信息：\n$response,body数据：\n${JSONUtil.toJsonStr(results.asJava)}")
          }
        }
      }
    }
    results.+=(value)
  }


  /**
   * 做快照时执行，清空状态内数据，然后将算子内数据添加到状态
   */
  override def snapshotState(context: FunctionSnapshotContext): Unit = {
    resultsState.clear()
    results.foreach(r => resultsState.add(r))
  }

  /**
   * 初始化checkpoint，将状态内数据恢复到算子内
   */
  override def initializeState(context: FunctionInitializationContext): Unit = {
    val resultStateDescriptor: ListStateDescriptor[RateResult] = new ListStateDescriptor[RateResult]("上批数据缓存", TypeInformation.of(new TypeHint[RateResult] {}))
    resultsState = context.getOperatorStateStore.getListState(resultStateDescriptor)
    if (context.isRestored) {
      if (resultsState.get() != null) {
        resultsState.get().forEach(r => {
          results.+=(r)
          timestamp = r.time
        })
      }
    }
  }


}
